/*
 * <p>$Id: //depot/MarketData/nds/main/src/main/java/com/ebs/md/nds/management/AbstractConnectionListener.java#2 $ </p>
 * <p>$Change: 75206 $ </p>
 * <p>$Author: chelck $ </p>
 * <p>$Date: 2009/07/16 $ </p>
 *
 * Copyright 2007 EBS Dealing Resources Inc.
 * All Rights Reserved
 */

package com.ebs.md.fxmonitor.management;

import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.management.MBeanRegistration;
import javax.management.MBeanServer;
import javax.management.Notification;
import javax.management.NotificationBroadcasterSupport;
import javax.management.ObjectName;

import org.apache.log4j.Logger;

import com.ebs.commons.util.SysDump;

import com.ebs.commons.comp.IParsedPayload;

public abstract class AbstractConnectionListener extends NotificationBroadcasterSupport
                                                 implements MessageListener, MBeanRegistration {
    private static final AtomicLong     s_seqNo         = new AtomicLong();
    private static final Logger         s_logger        = Logger.getLogger(SysDump.getCallerClass());

    private final AtomicBoolean  m_registered    = new AtomicBoolean(false);

    private ObjectName  m_objectName;

    public AbstractConnectionListener() {
        super();
    }

    //MBeanRegistration methods
    public void postDeregister() {
        // TODO Auto-generated method stub

    }

    public void postRegister(Boolean registrationDone) {
        // TODO Auto-generated method stub

    }

    public void preDeregister() throws Exception {
        setRegistered(false);
    }

    public ObjectName preRegister(MBeanServer server, ObjectName name) throws Exception {
        setRegistered(true);
        s_logger.info("AbstractConnectionListener.preRegister() ObjectName: " + name);
        m_objectName = name;
        return name;
    }

    //MessageListener methods
    public final void onMessage(Message message) {
        Notification notification = null;
        try {
            if(isRegistered()) {
                IParsedPayload payload = extractPayload(message);
                if(processPayload(payload)) {
                    notification = createNotification(payload);
                    sendNotification(notification);
                }
            }
            else {
   //             s_logger.warn("Attempt to Brodcast JMX Notification before MBean registration: " + extractPayload(message).toString());
            }
        }
        catch (JMSException ex) {
            s_logger.error("Unable to deciver JMSMessage: " + message, ex);
        }
    }

    protected boolean isRegistered() {
        return m_registered.get();
    }

    protected void setRegistered(boolean state) {
        m_registered.set(state);
    }

    protected ObjectName getObjectName() {
        return m_objectName;
    }

    protected final long getNextSeqNo() {
        return s_seqNo.addAndGet(1);
    }

    protected IParsedPayload extractPayload(Message message) throws JMSException {
        IParsedPayload payload = null;

        if(message instanceof ObjectMessage) {
            ObjectMessage   objMsg  = (ObjectMessage) message;
            payload = (IParsedPayload) objMsg.getObject();
        }

        return payload;
    }

    protected abstract boolean processPayload(IParsedPayload payload);
    protected abstract Notification createNotification(IParsedPayload payload) throws JMSException;

}
